package com.github.prontera.service;

import java.text.MessageFormat;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.task.TaskRejectedException;
import org.springframework.dao.DataAccessException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import com.github.prontera.domain.Sequence;
import com.github.prontera.persistence.CrudMapper;
import com.github.prontera.persistence.SequenceMapper;
import com.github.prontera.service.CrudServiceImpl;

/**
 * 生产序列号接口
 */
@Service
public class SequenceService extends CrudServiceImpl<Sequence>{

	public SequenceService(CrudMapper<Sequence> mapper) {
		super(mapper);
	}

	private static final Logger logger = LoggerFactory.getLogger(SequenceService.class);

	@Autowired
	private SequenceMapper sequenceMapper;

	/**
	 * 异步刷新线程池
	 */
	@Autowired
	@Qualifier("sequenceFlushTreadPoolTaskExecutor")
	private ThreadPoolTaskExecutor ThreadPoolTaskExecutor;

	/**
	 * 缓存队列
	 */
	private ConcurrentHashMap<String, ConcurrentLinkedQueue<Long>> sequenceQueue = new ConcurrentHashMap<String, ConcurrentLinkedQueue<Long>>();

	/**
	 * 刷新锁
	 */
	private ConcurrentHashMap<String, ReentrantLock> locks = new ConcurrentHashMap<String, ReentrantLock>();

	/**
	 * 计数器
	 */
	private ConcurrentHashMap<String, AtomicInteger> counts = new ConcurrentHashMap<String, AtomicInteger>();

	/**
	 * 阀值
	 */
	private Map<String, Integer> thresholds = new HashMap<String, Integer>();

	/**
	 * 总数
	 */
	private Map<String, Integer> totals = new HashMap<String, Integer>();

	@PostConstruct
	public void init()  {
		logger.info("初始化序列号...");
		if (sequenceQueue.isEmpty()) {
			List<Sequence> sequenceList = sequenceMapper.find(new Sequence());
			for (Sequence sequence : sequenceList) {
				this.thresholds.put(sequence.getName(), sequence.getThreshold());
				this.totals.put(sequence.getName(), sequence.getTotal());
				// 初始化锁队列
				this.locks.put(sequence.getName(), new ReentrantLock());
				// 更新缓冲区
				flushBuffer(sequence.getName());
			}
		}
		logger.info("初始化序列号完成...");
	}

	/**
	 * 刷新缓冲区
	 */
	@Transactional(propagation = Propagation.REQUIRED)
	public void flushBuffer(final String sequenceName) {
			Sequence sequence = sequenceMapper.lock(sequenceName);
			if (sequence == null) {
				logger.error("序列[{}]没有初始化.", sequenceName);
				throw new RuntimeException("序列" + sequenceName + "没有初始化.");
			}
			if (SequenceService.this.thresholds.get(sequence.getName()) == null) {
				SequenceService.this.thresholds.put(sequence.getName(), sequence.getThreshold());
			}
			if (SequenceService.this.totals.get(sequence.getName()) == null) {
				SequenceService.this.totals.put(sequence.getName(), sequence.getTotal());
			}
			if (SequenceService.this.locks.get(sequence.getName()) == null) {
				SequenceService.this.locks.put(sequence.getName(), new ReentrantLock());
			}
			int increment = sequence.getIncrement().intValue();
			int total = sequence.getTotal().intValue();
			Integer offset = Integer.valueOf(increment * total);
			Long maxValue = Long.valueOf((sequence.getMaxValue() == null) || (sequence.getMaxValue().longValue() <= 0L) ? Long.MAX_VALUE : sequence
					.getMaxValue().longValue());
			// 计算更新值
			Long beforeValue = sequence.getCurrentValue();
			Long afterValue = beforeValue + offset;
			while (afterValue > maxValue) {
				afterValue = afterValue - maxValue;
			}
			// 更新
			try {
				sequenceMapper.updateForValue(sequenceName, beforeValue, afterValue);
			} catch (DataAccessException e) {
				logger.error(MessageFormat.format("刷新缓冲区更新序列号时出错，sequenceName:{},beforeValue:{},afterValue:{}", sequenceName, beforeValue, afterValue, e));
				throw e;
			}
			// 创建新的队列
			ConcurrentLinkedQueue<Long> newQueue = new ConcurrentLinkedQueue<Long>();
			for (int i = 0; i < total; i++) {
				newQueue.add(beforeValue);
				beforeValue += increment;
				if (beforeValue > maxValue) {
					beforeValue = beforeValue - maxValue;
				}
			}
			// 清空数据，初始化
			sequenceQueue.put(sequenceName, newQueue);
			// 设置计数器
			counts.put(sequenceName, new AtomicInteger(0));
	}

	public void flush(String sequenceName) {
		// 检查阀值
		if (overThreshold(sequenceName)) {
			// 尝试获取锁
			// 若返回false则已被锁定，当前线程不做操作;如果返回true则可以拿到当前线程锁
			try {
				locks.get(sequenceName).lock();
				// 锁定后再次检测能否获取锁
				if (overThreshold(sequenceName)) {
					try {
						flushBuffer(sequenceName);
					} catch (DataAccessException e) {
						// 更新异常则通过线程池异步更新
						asyncFlush(sequenceName);
					}
				}
			} finally {
				// 释放锁
				locks.get(sequenceName).unlock();
			}
		}
	}

	/**
	 * 获取序列号
	 */
	public Long next(String sequenceName) {
		ConcurrentLinkedQueue<Long> queue = sequenceQueue.get(sequenceName);
		if (queue == null) {
			flushBuffer(sequenceName);
			queue = sequenceQueue.get(sequenceName);
		}
		Long sequence = queue.poll();
		if (sequence != null) {
			// 计数增长
			counts.get(sequenceName).incrementAndGet();
			// 若达到刷新阀值,异步刷新
			if (overThreshold(sequenceName)) {
				asyncFlush(sequenceName);
			}
			return sequence;
		}
		// 若队列取空
		try {
			// 立即刷新
			flush(sequenceName);
			// 重新抓取
			return next(sequenceName);
		} catch (Exception e) {
			logger.error("获取序列号失败, 重新获取.[{}]", sequenceName, e);
			return next(sequenceName);
		}
	}

	/**
	 * 是否越过刷新阀值
	 * 
	 * @param sequenceName
	 * @return
	 */
	private boolean overThreshold(String sequenceName) {
		return counts.get(sequenceName).intValue() >= totals.get(sequenceName) - thresholds.get(sequenceName);
	}

	/**
	 * 异步刷新
	 * 
	 * @param sequenceName
	 */
	private void asyncFlush(final String sequenceName) {
		try {
			ThreadPoolTaskExecutor.execute(new Runnable() {
				@Override
				public void run() {
					flush(sequenceName);
				}
			});
		} catch (TaskRejectedException e) {
		}
	}


}
